package com.taobao.ad.hadoop.common;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapred.*;

public abstract class JobNode {
	public JobConf getJobConf() {
		return jobConf;
	} 
	public void set(String name, String value) {
		kvProperties.put(name, value);
	} 
	public String get(String name) {
		return kvProperties.get(name);
	} 
	public abstract void ConfigJob();

	public void ConfigJob(Class<?> MainClass,
			Class<? extends Mapper<?, ?, ?, ?>> MapperClass,
			Class<? extends Reducer<?, ?, ?, ?>> ReducerClass,
			Class<? extends Reducer<?, ?, ?, ?>> CombinerClass) {
		jobConf = new JobConf(MainClass);
		for (String propertyKey : this.kvProperties.keySet()) {
			jobConf.set(propertyKey, this.kvProperties.get(propertyKey));
		}
		FileInputFormat.setInputPaths(jobConf, this.get("inputPath"));
		FileOutputFormat.setOutputPath(jobConf,new Path(this.get("outputPath")));
		jobConf.setNumMapTasks(Integer.parseInt(this.get("nmap")));
		jobConf.setNumReduceTasks(Integer.parseInt(this.get("nreduce")));
		jobConf.setMapperClass(MapperClass);
		jobConf.setReducerClass(ReducerClass);
		if (CombinerClass != null) {
			jobConf.setCombinerClass(CombinerClass);
		}
		
		if (this.get("infmt").equals("text")) {
			jobConf.setInputFormat(TextInputFormat.class);
		} else if (this.get("infmt").equals("kvtext")) {
			jobConf.setInputFormat(KeyValueTextInputFormat.class);
		} else if (this.get("infmt").equals("seq")){
			jobConf.setInputFormat(SequenceFileInputFormat.class);
		} 
		
		if (this.get("outfmt").equals("text")) {
			jobConf.setOutputFormat(TextOutputFormat.class);
		} else {
			jobConf.setOutputFormat(SequenceFileOutputFormat.class);
			SequenceFileOutputFormat.setCompressOutput(jobConf, true);
			SequenceFileOutputFormat.setOutputCompressorClass(jobConf,GzipCodec.class);
			SequenceFileOutputFormat.setOutputCompressionType(jobConf,SequenceFile.CompressionType.BLOCK);
		}
		jobConf.setMapOutputKeyClass(Text.class);
		jobConf.setMapOutputValueClass(Text.class);
		jobConf.setOutputKeyClass(Text.class);
		jobConf.setOutputValueClass(Text.class);
	}

	public boolean runJob(int jobindex) {
		int curjobindex = Integer.parseInt(this.get("jobindex"));
		if (curjobindex >= jobindex) {
			try {
				this.PrintArgs();
				FileSystem hdfs = FileSystem.get(new Configuration());
				hdfs.delete(new Path(this.get("outputPath")), true);
				return JobClient.runJob(jobConf).isSuccessful();
			} catch (IOException e) {
				e.printStackTrace();
				return false;
			}
		} else {
			this.PrintArgs();
			return true;
		}
	}

	private void PrintArgs() {
		System.out.println("\n-----------------------------------------------------------------------\n");
		System.out.println("[jobindex]:" + this.get("jobindex"));
		System.out.println("[jobname]:" + this.get("jobname"));
		System.out.println("[input]:" + this.get("inputPath"));
		System.out.println("[output]:" + this.get("outputPath"));
		System.out.println("[nmap]:" + this.get("nmap"));
		System.out.println("[nreduce]:" + this.get("nreduce"));
	}

	private JobConf jobConf;
	Map<String, String> kvProperties = new HashMap<String, String>();
}
